package my.twitter.client;

import static org.apache.http.HttpHeaders.CONTENT_TYPE;
import static org.apache.http.HttpStatus.SC_MULTIPLE_CHOICES;
import static org.apache.http.HttpStatus.SC_OK;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;

import java.net.URI;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import oauth.signpost.exception.OAuthException;

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.DefaultHttpAsyncClient;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.util.EntityUtils;
import org.json.JSONException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

public class TwitterHttpPost extends DefaultHttpAsyncClient {

    private final Logger log = LoggerFactory.getLogger(getClass());

    @Autowired
    private TwitterUser user;

    @Autowired
    private TwitterRequest request;

    private URI filterUri;

    private int timeout;

    public TwitterHttpPost() throws IOReactorException {
	super();

	getParams().setBooleanParameter(CoreConnectionPNames.SO_KEEPALIVE, true)
		.setBooleanParameter(CoreConnectionPNames.SO_REUSEADDR, true);
    }

    public URI getFilterUri() {
	return filterUri;
    }

    public void setFilterUri(URI filterUri) {
	this.filterUri = filterUri;
    }

    public int getTimeout() {
	return timeout;
    }

    public void setTimeout(int timeout) {
	this.timeout = timeout;

	getParams().setIntParameter(CoreConnectionPNames.SO_TIMEOUT, timeout).setIntParameter(
		CoreConnectionPNames.CONNECTION_TIMEOUT, timeout);
    }

    public void executeStatusesFilter() {
	start();

	final CountDownLatch latch = new CountDownLatch(1);

	try {
	    String jsonRequest = request.getRequestBody();
	    log.info("Request body: " + jsonRequest);

	    final HttpPost httpRequest = new HttpPost(filterUri);
	    httpRequest.addHeader(CONTENT_TYPE, APPLICATION_JSON_VALUE);
	    // httpRequest.addHeader(ACCEPT, APPLICATION_JSON_VALUE);
	    httpRequest.setEntity(new StringEntity(jsonRequest, ContentType.APPLICATION_JSON));

	    user.sign(httpRequest);

	    execute(httpRequest, new FutureCallback<HttpResponse>() {
		public void completed(final HttpResponse response) {
		    int statusCode = response.getStatusLine().getStatusCode();
		    if (statusCode < SC_OK || statusCode >= SC_MULTIPLE_CHOICES) {
			log.warn("HTTP " + statusCode + " -> "
				+ response.getStatusLine().getReasonPhrase());
			latch.countDown();
			return;
		    }

		    try {
			log.info("Tweet: " + EntityUtils.toString(response.getEntity()));
		    } catch (Exception e) {
			log.error("Reading content of HttpResponse:", e);
			latch.countDown();
		    }
		}

		public void failed(final Exception e) {
		    log.error("HTTP connection failed:", e);
		    latch.countDown();
		}

		public void cancelled() {
		    log.warn("HTTP connection cancelled");
		    latch.countDown();
		}
	    });

	    latch.await(timeout, TimeUnit.MILLISECONDS);
	} catch (JSONException e) {
	    log.error("Invalid HTTP Request:", e);
	} catch (OAuthException e) {
	    log.error("Signing HTTP Request:", e);
	} catch (InterruptedException e) {
	    log.error("Waiting for completion:", e);
	} finally {
	    try {
		shutdown();
	    } catch (InterruptedException e) {
		log.warn("Shutdown HTTP connection:", e);
	    }
	}
    }
}
